// 发布订阅管理器

import { BaseManager } from '../core/BaseManager';
import { RedisClient } from '../core/RedisClient';
import { KeyGenerator } from '../utils/keyGenerator';
import { EventEmitter } from 'events';

export interface PubSubMessage {
  channel: string;
  pattern?: string;
  message: any;
  timestamp: Date;
}

export interface PubSubStats {
  publishedMessages: number;
  subscribedChannels: number;
  patternSubscriptions: number;
  activeSubscribers: number;
}

export interface SubscriptionOptions {
  pattern?: boolean;
  autoReconnect?: boolean;
  maxRetries?: number;
  retryDelay?: number;
}

export class PubSubManager extends BaseManager {
  private eventEmitter: EventEmitter;
  private subscriptions: Map<string, SubscriptionOptions>;
  private patternSubscriptions: Map<string, SubscriptionOptions>;
  private subscriberClient?: any;
  private stats: PubSubStats;

  constructor(client: RedisClient) {
    super(client, new KeyGenerator({ prefix: 'pubsub' }));
    this.eventEmitter = new EventEmitter();
    this.subscriptions = new Map();
    this.patternSubscriptions = new Map();
    this.stats = {
      publishedMessages: 0,
      subscribedChannels: 0,
      patternSubscriptions: 0,
      activeSubscribers: 0,
    };
  }

  /**
   * 发布消息到频道
   */
  async publish(channel: string, message: any): Promise<number> {
    this.validateRequired({ channel, message });

    const startTime = this.now();

    try {
      const serializedMessage = typeof message === 'string' ? message : JSON.stringify(message);

      const subscriberCount = await this.execute<number>('publish', channel, serializedMessage);

      this.stats.publishedMessages++;

      this.logMetric('pubsub.publish', this.now() - startTime, true, {
        channel,
        subscriberCount,
        messageSize: serializedMessage.length,
      });

      return subscriberCount;
    } catch (error) {
      this.logMetric('pubsub.publish', this.now() - startTime, false, {
        channel,
        error: (error as Error).message,
      });
      throw this.formatError(error, 'Failed to publish message');
    }
  }

  /**
   * 订阅频道
   */
  async subscribe(
    channel: string,
    callback: (message: PubSubMessage) => void,
    options: SubscriptionOptions = {}
  ): Promise<void> {
    this.validateRequired({ channel, callback });

    try {
      await this.ensureSubscriberClient();

      this.subscriptions.set(channel, options);
      this.stats.subscribedChannels++;

      // 设置消息处理器
      this.eventEmitter.on(`message:${channel}`, callback);

      await this.subscriberClient.subscribe(channel);

      this.logger.info(`Subscribed to channel: ${channel}`);
    } catch (error) {
      throw this.formatError(error, `Failed to subscribe to channel: ${channel}`);
    }
  }

  /**
   * 模式订阅
   */
  async psubscribe(
    pattern: string,
    callback: (message: PubSubMessage) => void,
    options: SubscriptionOptions = {}
  ): Promise<void> {
    this.validateRequired({ pattern, callback });

    try {
      await this.ensureSubscriberClient();

      this.patternSubscriptions.set(pattern, options);
      this.stats.patternSubscriptions++;

      // 设置模式消息处理器
      this.eventEmitter.on(`pmessage:${pattern}`, callback);

      await this.subscriberClient.psubscribe(pattern);

      this.logger.info(`Subscribed to pattern: ${pattern}`);
    } catch (error) {
      throw this.formatError(error, `Failed to subscribe to pattern: ${pattern}`);
    }
  }

  /**
   * 取消订阅频道
   */
  async unsubscribe(channel: string): Promise<void> {
    this.validateRequired({ channel });

    try {
      if (this.subscriberClient) {
        await this.subscriberClient.unsubscribe(channel);
      }

      this.subscriptions.delete(channel);
      this.eventEmitter.removeAllListeners(`message:${channel}`);

      if (this.stats.subscribedChannels > 0) {
        this.stats.subscribedChannels--;
      }

      this.logger.info(`Unsubscribed from channel: ${channel}`);
    } catch (error) {
      throw this.formatError(error, `Failed to unsubscribe from channel: ${channel}`);
    }
  }

  /**
   * 取消模式订阅
   */
  async punsubscribe(pattern: string): Promise<void> {
    this.validateRequired({ pattern });

    try {
      if (this.subscriberClient) {
        await this.subscriberClient.punsubscribe(pattern);
      }

      this.patternSubscriptions.delete(pattern);
      this.eventEmitter.removeAllListeners(`pmessage:${pattern}`);

      if (this.stats.patternSubscriptions > 0) {
        this.stats.patternSubscriptions--;
      }

      this.logger.info(`Unsubscribed from pattern: ${pattern}`);
    } catch (error) {
      throw this.formatError(error, `Failed to unsubscribe from pattern: ${pattern}`);
    }
  }

  /**
   * 获取频道的订阅者数量
   */
  async getSubscriberCount(channel: string): Promise<number> {
    this.validateRequired({ channel });

    try {
      const result = await this.execute<any>('pubsub', 'numsub', channel);
      return Array.isArray(result) && result.length > 1 ? parseInt(result[1], 10) : 0;
    } catch (error) {
      throw this.formatError(error, `Failed to get subscriber count for channel: ${channel}`);
    }
  }

  /**
   * 获取活跃频道列表
   */
  async getActiveChannels(pattern?: string): Promise<string[]> {
    try {
      const args = pattern ? ['channels', pattern] : ['channels'];
      return await this.execute<string[]>('pubsub', ...args);
    } catch (error) {
      throw this.formatError(error, 'Failed to get active channels');
    }
  }

  /**
   * 获取统计信息
   */
  getStats(): PubSubStats {
    return { ...this.stats };
  }

  /**
   * 清理所有订阅
   */
  async cleanup(): Promise<void> {
    try {
      if (this.subscriberClient) {
        // 取消所有订阅
        for (const channel of this.subscriptions.keys()) {
          await this.unsubscribe(channel);
        }

        for (const pattern of this.patternSubscriptions.keys()) {
          await this.punsubscribe(pattern);
        }

        // 断开订阅客户端
        await this.subscriberClient.disconnect();
        this.subscriberClient = undefined;
      }

      // 清理事件监听器
      this.eventEmitter.removeAllListeners();

      this.logger.info('PubSub cleanup completed');
    } catch (error) {
      throw this.formatError(error, 'Failed to cleanup PubSub');
    }
  }

  /**
   * 确保订阅客户端存在
   */
  private async ensureSubscriberClient(): Promise<void> {
    if (!this.subscriberClient) {
      // 简化实现：使用主客户端进行订阅
      // 在实际项目中，应该创建专用的订阅客户端
      this.subscriberClient = this.client;
      this.stats.activeSubscribers++;
    }
  }
}
